package com.example.order.mq;

import com.alibaba.fastjson.JSON;
import com.example.order.constant.ShopCode;
import com.example.order.entity.*;
import com.example.order.mapper.TradeGoodsMapper;
import com.example.order.mapper.TradeGoodsNumberLogMapper;
import com.example.order.mapper.TradeMqConsumerLogMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.Date;

@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING)
public class CancelMQListener implements RocketMQListener<MessageExt> {

    @Value("${mq.order.topic}")
    private String groupName;

    @Autowired
    private TradeGoodsMapper goodsMapper;

    @Autowired
    private TradeGoodsNumberLogMapper goodsNumberLogMapper;

    @Autowired
    private TradeMqConsumerLogMapper mqConsumerLogMapper;


    @Override
    public void onMessage(MessageExt messageExt) {
        try {
            //        1.解析消息内容
            log.info("解析消息内容");
            String msgId = messageExt.getMsgId();
            String tags = messageExt.getTags();
            String keys = messageExt.getKeys();
            String body =new String(messageExt.getBody(),"UTF-8");
            //         2.查询消息消费记录
            log.info("查询消息消费记录");
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(groupName,tags,keys);
            if(mqConsumerLog!=null){
//                3.  判断如果消费过...
//                3.1 获得消息处理状态
                Integer status = mqConsumerLog.getConsumerStatus();
//                处理过...返回
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCESS.getCode().intValue()==status.intValue()){
                    log.info("消息:"+msgId+",已经处理了");
                    return;
                }

//                正在处理
                if(ShopCode.SHOP_MQ_MESSAGE_STATUT_PROCESSING.getCode().intValue()==status.intValue()){
                    log.info("消息："+msgId+",正在处理");
                }
//                处理失败
                if(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue()==status.intValue()){
                    Integer times = mqConsumerLog.getConsumerTimes();
                    if(times>3){
                        log.info("消息："+msgId+",处理失败3次");
                    }
                    else {
//                        如果没有超过三次，使用乐观锁来继续处理
                        mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUT_PROCESSING.getCode());
                        TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
                        TradeMqConsumerLogExample.Criteria criteria = example.createCriteria();
                        criteria.andMsgTagEqualTo(mqConsumerLog.getMsgTag());
                        criteria.andMsgKeyEqualTo(mqConsumerLog.getMsgKey());
                        criteria.andGroupNameEqualTo(mqConsumerLog.getGroupName());
                        criteria.andConsumerTimesEqualTo(mqConsumerLog.getConsumerTimes());
                        int r = mqConsumerLogMapper.updateByExampleSelective(mqConsumerLog,example);
                        if(r<=0){
                            log.info("并发处理");
                        }
                    }
                }
                else {
//                    4判断如果没有消费过
                    log.info("首次消费此消息");
                    mqConsumerLog = new TradeMqConsumerLog();
                    mqConsumerLog.setMsgTag(tags);
                    mqConsumerLog.setMsgKey(keys);
                    mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUT_PROCESSING.getCode());
                    mqConsumerLog.setMsgBody(body);
                    mqConsumerLog.setMsgId(msgId);
                    mqConsumerLog.setConsumerTimes(0);
//                    将消息处理信息添加到数据库
                    mqConsumerLogMapper.insert(mqConsumerLog);
                }
            }
            //        5.回退库存
            MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
            Long goodsId = mqEntity.getGoodsId();
            TradeGoods goods = goodsMapper.selectByPrimaryKey(mqEntity.getGoodsId());
            goods.setGoodsNumber(goods.getGoodsNumber()+mqEntity.getGoodsNum());
            goodsMapper.updateByPrimaryKey(goods);
            log.info("回退库存成功");
//        6.记录消息消费日志
            TradeGoodsNumberLog goodsNumberLog = new TradeGoodsNumberLog();
            goodsNumberLog.setOrderId(mqEntity.getOrderId());
            goodsNumberLog.setGoodsId(goodsId);
            goodsNumberLog.setGoodsNumber(mqEntity.getGoodsNum());
            goodsNumberLog.setLogTime(new Date());
            goodsNumberLogMapper.insert(goodsNumberLog);
            log.info("记录消息消费日志");
//          7.消息状态改为处理成功
            mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCESS.getCode());
            mqConsumerLog.setConsumerTimestamp(new Date());
            mqConsumerLogMapper.updateByPrimaryKey(mqConsumerLog);
            log.info("消息状态改为处理成功");
            log.info("回退库存正常结束");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }

    }
}
